package com.aotain.nyx.icmpflood;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.flink.BasicTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

/**
 * 
        *@Title:ICMP Flood 检测主方法
        *@Description:
        *@Author:Turk
        *@Since:2017年6月26日
        *@Version:1.1.0
 */
public class ICMPFloodDetect implements FlatMapFunction<Tuple2<String,BasicTuple>, AbnStatisTuple>
{

	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctets = new HashMap<String,Long>();
	
	/**
	 * 上行包数
	 */
	private HashMap<String,Long> mapStreampacket = new HashMap<String,Long>();
	
	/**
	 * 存放记录连续状态 最后的更新时间
	 */
	private HashMap<String,Long> mapLastTime = new HashMap<String,Long>();
	
	/**
	 * 记录攻击开始时间
	 */
	private HashMap<String,Long> mapAttackStart = new HashMap<String,Long>();
	
	//-----累积包，次数
	/**
	 * 持久-上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctetsMem = new HashMap<String,Long>();
		
	/**
	 * 持久-上行包数
	 */
	private HashMap<String,Long> mapStreampacketMem = new HashMap<String,Long>();
		
	/**
	 * 持续攻击时记录第一次开始的Map拼接的EntryKey
	 */
	private HashMap<String,String> mapEntryKey = new HashMap<String,String>();
	
	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();
	
	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;
	
	/**
	 * 发包个数，门限 30个/秒
	 */
	private final long ATTACK_THRESHOLD = 30*60;
	
	/**
	  *
	  */
	    
	private static final long serialVersionUID = 7016107096096488299L;

	@Override
	public void flatMap(Tuple2<String, BasicTuple> value,
			Collector<AbnStatisTuple> out) throws Exception {
		BasicTuple tuple = value.f1;
		
		String hashKey = tuple.getDestIP() + "_" + tuple.getSourceIP();
		mapStreamoctets.put(hashKey, tuple.getUpStreamOctets());
		mapStreampacket.put(hashKey, tuple.getUpStreamPacket());
		
		if(tuple.getUpStreamPacket() >= ATTACK_THRESHOLD) {
			//判断当前源->目的IP是否存在记录
			if(mapLastTime.containsKey(hashKey))
			{
				//存在记录，更新数据
				mapLastTime.put(hashKey, System.currentTimeMillis());
				Logger.getLogger(ICMPFloodDetect.class).info("###ICMP ATTACK UPDATE" + hashKey);
				AbnStatisTuple abnTuple = UpdateAttackLog(hashKey,tuple);
				out.collect(abnTuple);
			}
			else
			{
				//新发生的攻击行为，记录开始时间并且上报异常攻击日志，但是不上报统计数据
				mapLastTime.put(hashKey, System.currentTimeMillis());
				AbnStatisTuple abnTuple = InsertNewAttackLog(hashKey, tuple);
				out.collect(abnTuple);
				Logger.getLogger(ICMPFloodDetect.class).info("###ICMP ATTACK START" + hashKey);
			}
		} else {
			
			if(mapLastTime.containsKey(hashKey))
			{
				//如果存在记录，判断更新时间与当前时间差距是否超过了等待门限
				Long lastTime = mapLastTime.get(hashKey);
				if(System.currentTimeMillis() - lastTime > END_DURATION)
				{
					//超过30秒，认为攻击结束，记录结束时间，
					Logger.getLogger(ICMPFloodDetect.class).info("###ICMP ATTACK END" + hashKey);
					EndAttackLog(hashKey,tuple);
					//将数据从map中移除
					//mapWUrl.remove(base64Key);
					mapLastTime.remove(hashKey);
					mapStreamoctetsMem.remove(hashKey);
					mapStreampacketMem.remove(hashKey);
					mapAttackStart.remove(hashKey);
					mapEntryKey.remove(hashKey);
					mapHighFlow.remove(hashKey);
					//Logger.getLogger(CCDetectBoltV2.class).info("###CC V2 ATTACK END" + entryKey);
				} else 	{
					//虽然未超过门限，但是在攻击等待范围时间内，将这一分钟的数据累加到攻击数据中。
					AbnStatisTuple abnTuple = UpdateAttackLog(hashKey,tuple);
					out.collect(abnTuple);
				}
			}
		}
		
		// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束
		ArrayList<String> removeItem = new ArrayList<String>();
		for(Entry<String, Long> entry : mapLastTime.entrySet()) {
			Long lastTime = entry.getValue();
			String base64Key = entry.getKey();
			Logger.getLogger(ICMPFloodDetect.class).info(
			String.format("###ICMPFlood REMOVE hashkey %s time %d",base64Key,(System.currentTimeMillis() - lastTime)/1000));
						
			if(System.currentTimeMillis() - lastTime > END_DURATION) {
				//超过等待门限秒，认为攻击结束，记录结束时间，
				String entryKey = mapEntryKey.get(base64Key);
				Logger.getLogger(ICMPFloodDetect.class).info("###ICMPFlood ATTACK END" + entryKey);
				EndAttackLog(entryKey,tuple);
				//将数据从map中移除
				//mapWUrl.remove(base64Key);
				removeItem.add(base64Key);
				mapStreamoctetsMem.remove(base64Key);
				mapStreampacketMem.remove(base64Key);
				mapAttackStart.remove(base64Key);
				mapEntryKey.remove(base64Key);
				mapHighFlow.remove(base64Key);
							
			}
		}
					
		//remove 已经攻击结束的记录
		for(String base64Key : removeItem) {
			mapLastTime.remove(base64Key);
		}
					
		removeItem.clear();
		mapStreamoctets.clear();
		mapStreampacket.clear();
	}
	
	
	private AbnStatisTuple InsertNewAttackLog(String tupleKey,BasicTuple tuple)
	{		
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		
		String gis = new String(Base64.decodeBase64(tuple.getGIS()));
		String[] gisArray = gis.split("#",-1);
		
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];					
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		
		long starttime = System.currentTimeMillis();
		mapAttackStart.put(tupleKey,starttime);//记录开始时间
		
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(starttime));
		String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "12";	

		String desc = String.format("ICMP Flood 攻击 ");
		
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("ROWKEY",rowKeyAbnormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP",sip));
		row.add(new DataColumn("DESTPORT",""));
		row.add(new DataColumn("ACCESSTIME",Long.parseLong(accesstime)));
		row.add(new DataColumn("ABRNORMAL", 12));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", tuple.getUpStreamPacket()));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("PROVINCE", sourceProvinceName));
		row.add(new DataColumn("FLOWDIRECTION", tuple.getFlowDirection()));
		
		
		
		//UDII数据特有
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		
		
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctets.get(tupleKey);
			nUpstreampacket = mapStreampacket.get(tupleKey);
			row.add(new DataColumn("UPSTREAMOCTETS", nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET", nUpstreampacket));
			
			mapStreamoctetsMem.put(tupleKey, nUpStreamOctets);
			mapStreampacketMem.put(tupleKey, nUpstreampacket);
		}
		
		importtool.InsertRowData(tableName, row);
		
		mapEntryKey.put(tupleKey, tupleKey);
		
		Logger.getLogger(ICMPFloodDetect.class).info(
				String.format("###ICMP Flood ATTACK START ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, tupleKey));
		
		double maxValue =  nUpStreamOctets/(1024*60);
		mapHighFlow.put(tupleKey, maxValue);
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("ICMPFLOOD");
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamPacket(nUpstreampacket);
		abnTuple.setAttackNum(1l);
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("1");
		abnTuple.setAttTime(accesstime);
		abnTuple.setAttackMax(maxValue);
		return abnTuple;
	}
	
	
	private AbnStatisTuple UpdateAttackLog(String tupleKey,BasicTuple tuple)
	{
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		
		String gis = new String(Base64.decodeBase64(tuple.getGIS()));
		String[] gisArray = gis.split("#",-1);
		
		String sourceAreaCountry = gisArray[4];					
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		//UDII数据特有
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctets.get(tupleKey);
			nUpstreampacket = mapStreampacket.get(tupleKey);
			if(mapStreamoctetsMem.containsKey(tupleKey))
				mapStreamoctetsMem.put(tupleKey, mapStreamoctetsMem.get(tupleKey) + nUpStreamOctets);
			if(mapStreamoctetsMem.containsKey(tupleKey))
				mapStreampacketMem.put(tupleKey, mapStreamoctetsMem.get(tupleKey) + nUpstreampacket);
		}
		
		String desc = String.format("ICMP Flood 攻击 ");
		double maxValue = (double)nUpStreamOctets/(double)60/(double)1024 ;
		
		if(mapHighFlow.containsKey(tupleKey))
		{
			double maxValueMem = mapHighFlow.get(tupleKey);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(tupleKey, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(tupleKey, maxValue);
		}
		
		desc = String.format("%s 攻击峰值为 :%f KB/s ", desc,maxValue);
		
	
		
		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(tupleKey)));
				
		String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "12";
				
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(tupleKey);
			nUpstreampacket = mapStreampacketMem.get(tupleKey);
			row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
			
			desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		}
		
		
		desc = String.format("%s 攻击次数 [%d]",desc, nUpstreampacket);
		Logger.getLogger(ICMPFloodDetect.class).info(
				String.format("###ICMP V2 ATTACK UPDATE ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, tupleKey));

		row.add(new DataColumn("DESC",desc));
		row.add(new DataColumn("ATTNUM",nUpstreampacket));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("ICMPFLOOD");
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamPacket(nUpstreampacket);
		abnTuple.setAttackNum(0l);
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("1");
		abnTuple.setAttackMax(maxValue);
		abnTuple.setAttTime(accesstime);
		return abnTuple;
	}
	
	
	private void EndAttackLog(String tupleKey,BasicTuple tuple) {
		
		String dip = tuple.getDestIP();
		String sip = tuple.getSourceIP();
		
		
		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(tupleKey)));
		String endtime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapLastTime.get(tupleKey)+60*1000l));
		
		String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "12";
		
		
		String desc = String.format("ICMP Flood 攻击 ");
		
		
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(tupleKey);
			nUpstreampacket = mapStreampacketMem.get(tupleKey);
			row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
			
			desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		}


		row.add(new DataColumn("ATTNUM", nUpstreampacket));
		row.add(new DataColumn("ENDTIME", Long.parseLong(endtime)));
		
		
		desc = String.format("%s 攻击次数 [%d]",desc, nUpstreampacket);
		
		if(mapHighFlow.containsKey(tupleKey))
		{
			double maxValueMem = mapHighFlow.get(tupleKey);
			desc = String.format("%s 攻击峰值 [%f KB/s]",desc, maxValueMem);
		}
		row.add(new DataColumn("DESC",desc));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		
//		Date current = new Date(mapAttackStart.get(base64Key));
//		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
//		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
//		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
//		String dateStr_h = sdf_h.format(current);
//		String dateStr_d = sdf_d.format(current);
//		String dateStr_hour = sdf_hour.format(current);
		
		Logger.getLogger(ICMPFloodDetect.class).info(
				String.format("###ICMPFlood ATTACK END "
						+ "ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, tupleKey));
		
//		collector.emit(new Values(dip,sip,"","",dateStr_d, dateStr_h, sourceAreaId, sourceAreaName,
//				sourceProvinceName, countValue, dateStr_hour,"CC", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
		
	}
    
}

    